/**
 * Copyright ©2003~2019年5月6日 szkingdom
 * @author raohj
 * @date 2019年5月6日
 * @file RBlockQueue.java
 * @comment : please input your comment to this file
 * @version : input version
 * @modify:
 * 		1、 2019年5月6日 由 raohj 新建文件
 */
package com.ccnu.demojava.concurrent.tools;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.Random;
import java.util.concurrent.*;

/**
 * @author raohj
 * @date 2019年5月6日
 * @version
 * @comment 使用经典的生产者/消费者模式测试同步队列
 * 
 *          ArrayBlockingQueue : 数组队列 LinkedBlockingQueue : 链表队列
 */
public class RExchanger {
	private static final Logger log = LoggerFactory.getLogger(RExchanger.class);
	private static volatile boolean exitFlag = false;

	/**
	 * @author raohj
	 * @date 2019年5月6日
	 * @version
	 * @comment 生产者
	 */
	public static class RProduct implements Callable<Integer> {
		private Random rd = new Random(new Date().getTime());
		private List<String> msgQueue; // 消息队列
		private Exchanger<List<String>> ecg;

		/**
		 * 
		 */
		public RProduct(List<String> msgQueue, Exchanger<List<String>> ecg) {
			this.msgQueue = msgQueue;
			this.ecg = ecg;
		}

		@Override
		public Integer call() throws Exception {
			for (int r = 0; r < 2; r++) {
				for (int i = 0; i < 10; i++) {
					msgQueue.add("Msg" + rd.nextInt());
				}
				// 与消费者进行数据交换
				try {
					log.info("生产者：等待置换消息");
					this.msgQueue = ecg.exchange(this.msgQueue);
					log.info("生产者：消息置换成功");
				} catch (InterruptedException e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				}
			}
			return 0;
		}
	}

	/**
	 * @author raohj
	 * @date 2019年5月6日
	 * @version
	 * @comment 消费者
	 */
	public static class RConsumer implements Runnable {
		private List<String> msgQueue; // 消息队列
		private Exchanger<List<String>> ecg;

		/**
		 * 
		 */
		public RConsumer(List<String> msgQueue, Exchanger<List<String>> ecg) {
			this.msgQueue = msgQueue;
			this.ecg = ecg;
		}

		@Override
		public void run() {
			while (!exitFlag) {
				while (msgQueue.size() > 0) {
					log.info("消费者[" + Thread.currentThread().threadId() + "]:	" + msgQueue.get(0));
					msgQueue.remove(0);
					try {
						Thread.sleep(100);
					} catch (InterruptedException e) {
						e.printStackTrace();
					}
				}
				// 与生产者进行数据交换
				try {
					log.info("消费者：等待置换消息");
					this.msgQueue = ecg.exchange(this.msgQueue, 10, TimeUnit.SECONDS);
					log.info("消费者：消息置换成功");
				} catch (InterruptedException e) {
					e.printStackTrace();
				} catch (TimeoutException e) {
					log.info("消费者：消息置换超时");
				}
			}
		}
	}

	public static void test() {
		List<String> msg1 = new ArrayList<String>(); // 队列1
		List<String> msg2 = new ArrayList<String>(); // 队列2
		Exchanger<List<String>> ecg = new Exchanger<>();
		ExecutorService es = Executors.newFixedThreadPool(100);
		// 设置1个生产者
		Future<Integer> productRS = es.submit(new RProduct(msg1, ecg));
		es.submit(new RConsumer(msg2, ecg));
		
		try {
			if(productRS.get() == 0) {
				exitFlag = true;
			}
		} catch (InterruptedException | ExecutionException e) {
			e.printStackTrace();
		}
		es.shutdown();
	}
}
